// author : shike@outlook.com
//   date : 2013/07/11
//  brief : 定义VOIP TCP 主处理线程类型

#include "ct_voip_tcp_task.h"
#include "ct_voip_thread_pool.h"
#include "ctTime.h"
#include "ctLog.h"
#include <map>
#include <cassert>

class ChannelTCP
{
        public:
                ChannelTCP()
                {
                }
                ~ChannelTCP()
                {
                }
                void saveData(void *data, const DWORD &datalen, const DWORD &userid)
                {
                        if (0 == userid) {
                                for (std::map<DWORD, ctVOIPTCPTask *>::iterator it = task_map_.begin();
                                                it != task_map_.end();
                                                it++ ) {
                                        it->second->sendCmd(data, datalen, true);
                                }
                        }
                        else {
                                std::map<DWORD, ctVOIPTCPTask *>::iterator it = task_map_.find(userid);
                                if (task_map_.end() != it) {
                                        it->second->sendCmd(data, datalen, true);
                                }
                        }
                }
                void addTask(ctVOIPTCPTask *task)
                {
                        task_map_[task->getUserID()] = task;
                        Zebra::logger->info("[tcp/okay thread]total task num:%lu", task_map_.size());
                }
                void delTask(const DWORD &userid)
                {
                        std::map<DWORD, ctVOIPTCPTask *>::iterator it = task_map_.find(userid);
                        if (it != task_map_.end()) {
                                it->second->TerminateWait();
                                task_map_.erase(userid);
                        }
                        Zebra::logger->info("[tcp/okay thread]total task num:%lu", task_map_.size());
                }

                ctVOIPTCPTask *getTask(uint32_t user_id)
                {
                        std::map<DWORD, ctVOIPTCPTask *>::iterator it = task_map_.find(user_id);
                        if (it != task_map_.end())
                                return it->second;
                        else
                                return 0;
                }
        private:
                std::map<DWORD, ctVOIPTCPTask *> task_map_;
};

class ctVOIPTCPOkayThread : 
        public ctThread,
        public ctVOIPTCPTaskQueue
{
        private:

                ctVOIPThreadPool *pool;
                ctTCPTaskContainer tasks;                   /**< 任务列表 */
                ctTCPTaskContainer::size_type task_count;   /**< tasks计数(保证线程安全*/
                epollfdContainer epfds;
                int kdpfd;

                void _add(ctVOIPTCPTask *task)
                {
                        if (NULL == task) {
                                Zebra::logger->warn("[tcp/okay thread]%s, task is null", __PRETTY_FUNCTION__);
                                return ;
                        }
                        task->addEpollFD(kdpfd,EPOLLIN | EPOLLPRI |EPOLLERR | EPOLLOUT, (void*)task);

                        task->setOkayThread(this);
                        tasks.push_back(task);
                        task_count = tasks.size();
                        if(task_count > epfds.size()) {
                                epfds.resize(task_count+ 16);
                        }
                        task->ListeningRecv(false);
                }

                void remove(ctTCPTask_IT &it)
                {
                        ctVOIPTCPTask *task = *it;
                        task->delEpollFD(kdpfd,EPOLLIN | EPOLLPRI |EPOLLERR);
                        it = tasks.erase(it);
                        task_count = tasks.size();
                }

                t_BufferCmdQueue rcv_queue_rtp_;
                t_BufferCmdQueue rcv_queue_rtcp_;
                t_BufferCmdQueue snd_queue_temp_;    ctMutex mutex_snd_queue_;
                t_BufferCmdQueue snd_queue_;

        public:
                static const ctTCPTaskContainer::size_type connPerThread = 256;  /**< 每个线程带的连接数量 */

                ctVOIPTCPOkayThread(
                                ctVOIPThreadPool *pool,
                                const std::string &name = std::string("voip-tcp-okay-thread"))
                        : ctThread(name),pool(pool)
                {
                        task_count = 0;
                        kdpfd = epoll_create(connPerThread);
                        assert(-1 != kdpfd);
                        epfds.resize(connPerThread );
                        reset_data_ = false;
                }

                ~ctVOIPTCPOkayThread()
                {
                        ::close(kdpfd);
                }

                void run();

                inline const ctTCPTaskContainer::size_type size() const
                {
                        return task_count + _size;
                }

                void recvData(t_BufferCmdQueue &cmdqueue, eUDPProtoType type);
                void sendData();
                void saveUserData(t_BufferCmdQueue &data, eUDPProtoType type);
                void copyDataToRelay();
                void copyDataFromRelay(t_BufferCmdQueue &data);
                void processData();

                void addChannel(const DWORD &channelid);
                void delChannel(const DWORD &channelid);
                ChannelTCP *getChannel(const DWORD &channelid);
                void addTaskToChannel(const DWORD &channelid, ctVOIPTCPTask *task);
                void delTaskFromChannel(const DWORD &channelid, const DWORD &userid);
                void clearChannels();
                void setResetData();
                void resetData();

        private:
                std::map<DWORD/*channel*/, ChannelTCP *> channel_map_;

                bool reset_data_;
};


